package rpc_client_server;

/******************************************
Shaoni Sinha (shaoni@uw.edu)
Description: Server side code
******************************************/

import java.io.File;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.sql.Timestamp;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

//@SuppressWarnings("serial")
public class Server implements RemoteKeyValue {
	
	/** Number of threads in the server threadpool. */
	private static final int NTHREADS = 10;
	public static int NUMOFSERVERS = 5;
	private static double FAILURE_PROBABILITY = 0.5;
	
	public String name;
	
	public static HashMap<String, String> kvStore; /* The key-value store */
	public static File serverLog;
	public static Lock mutex;
	private ExecutorService executor;
	public int myPort;
	public int[] others;
	
	private int my_n;
	private int n_h;
	private int n_a;
	private Request req_a;
	private int highestN_a;
	private Request highestReq_a;
	
	
	/** List of pending requests of this server.*/
	private Map<UUID, Map<Integer, ACK>> firstACKMap;
	private Map<UUID, Map<Integer, ACK>> secondACKMap;
	
	public Server(final String theName, final int thePort) throws RemoteException {
		name = theName;
		executor = Executors.newFixedThreadPool(NTHREADS);
		kvStore = new HashMap<String, String>();
		serverLog = new File("serverlog.txt");
		mutex = new ReentrantLock(true);
		myPort = thePort;
		others = new int[NUMOFSERVERS-1];
		my_n = 0;
		n_h = 0;
		n_a = 0;
		req_a = null;
		highestN_a = 0;
		highestReq_a = null;
		
		firstACKMap = Collections.synchronizedMap(new HashMap<UUID,Map<Integer, ACK>>());
		secondACKMap = Collections.synchronizedMap(new HashMap<UUID,Map<Integer, ACK>>());
	}
	
	public String processRequestUsingThreadPool (Request req)
	{
		String response = null;
		MyCallable worker = new MyCallable(kvStore, req, mutex, serverLog, name);
		Future<String> submit = executor.submit(worker);
	    try {
			response = submit.get();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}
	    return response;
	}
	
	/** The method used to get the value corresponding to the key. */
	public String getValue(String key) throws RemoteException {
		System.out.println("" + new Timestamp(new java.util.Date().getTime()) + "  GET '" + key + "' request received in " + name);
		String [] args = {key};
		Request req = new Request("GET", args);
		return processRequestUsingThreadPool (req);
	}
	
	/** The method used to add a key-value pair. 
	 * @return 
	 * @throws NotBoundException */
	public String putKeyValue(String key, String value) throws RemoteException {
		System.out.println("" + new Timestamp(new java.util.Date().getTime()) + "  PUT ('" + key + "', '" + value + "') request received in " + name);
		UUID reqID = UUID.randomUUID();
		String req = "PUT";
		return this.threePhaseCommit(reqID, req, key, value);
	}
	
	/** The method used to delete a key-value pair. */
	public String deleteKeyValue(String key) throws RemoteException {
		System.out.println("" + new Timestamp(new java.util.Date().getTime()) + "  DELETE '" + key + "' request received in " + name);
		UUID reqID = UUID.randomUUID();
		String req = "DELETE";
		return this.threePhaseCommit(reqID, req, key, "");
		
	}

	private String threePhaseCommit(UUID reqId, String req, String key, String value) throws RemoteException{	
		
		String [] args = {key, value};
		req_a = new Request(reqId, req, args);
		
		while(true) 
		{
			/** Create a transaction id */
			UUID transactionId = UUID.randomUUID();
			
			/** Add request in my list. */
			this.addInRequestList(transactionId);
			
			/** Chooses my_n > n_h */
			my_n = ++n_h;
			
			/** Send prepare to others. */
			sendReqToOthers(transactionId);
			
			/** Wait for prepare ack from majority */
			boolean firstACKs = waitForAck(1, transactionId);
			if (!firstACKs)
			{
				System.out.print("" + new Timestamp(new java.util.Date().getTime()) + " " + this.name+ "  Failed to get Prepare acks from majority of servers! Restarting in 100ms.\r\n");
				/** Delay and restart paxos */
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				continue;
			}
			
			/** Leader gets promise (prepare-ok) from majority. So update the req_a. */
			if(n_h < highestN_a)
			{
				req_a = highestReq_a;
				n_a = highestN_a;
				n_h = n_a;
			}
			
			/** Send accept to others. */
			sendAcceptToOthers(transactionId);
			
			/** Wait for prepare ack from majority */
			boolean secondACKs = waitForAck(2, transactionId);
			if (!secondACKs)
			{
				System.out.print("" + new Timestamp(new java.util.Date().getTime()) + " " + this.name + "  Failed to get Accept acks from majority of servers!  Restarting in 100ms.\r\n");
				/** Delay and restart paxos */
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				continue;
			}
			
			/** Send commit to others. */
			sendCommitToOthers();
			
			return this.processRequestUsingThreadPool(req_a);
		}
	}
	
	private boolean waitForAck(int accNum, UUID reqId) 
	{
		int numOfTries = 0;
		while(numOfTries != 4){
			try 
			{
				Thread.sleep(100);
				boolean areMajorityAcked = false;
				if(accNum == 1)
				{
					areMajorityAcked = this.checkMajorityAcks(this.firstACKMap.get(reqId));
				} 
				else 
				{
					areMajorityAcked = this.checkMajorityAcks(this.secondACKMap.get(reqId));
				}
				if(areMajorityAcked)
				{
					return true;
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			numOfTries++;
		}
		return false;
	}
	
	private void sendReqToOthers(UUID transactionId) throws RemoteException
	{
		for(int i = 0; i< others.length; i++) 
		{
			Registry reg = LocateRegistry.getRegistry("localhost", others[i]);
			try 
			{
				 RemoteKeyValue stub = (RemoteKeyValue) reg.lookup(LOOKUPNAME);
				 stub.paxosHandlePrepareSignalInAcceptor(transactionId, my_n, myPort);
			} 
			catch (Exception e) 
			{
				e.printStackTrace();
				System.out.print("Failed to paxosSendPrepareSignalToAcceptor!");
			}
		}
	}
	
	private void sendAcceptToOthers(UUID transactionId) throws RemoteException
	{
		for(int i = 0 ; i<this.others.length; i++){
			int yourPort = others[i];
			Registry re = LocateRegistry.getRegistry("localhost", yourPort);
			RemoteKeyValue st;
			try {
				st = (RemoteKeyValue) re.lookup(LOOKUPNAME);
				st.paxosHandleAcceptSignalInAcceptor(transactionId, my_n, req_a.id, req_a.type, req_a.args[0], req_a.args[1], myPort);
			} catch (NotBoundException e) {
				e.printStackTrace();
			}
		}
	}
	
	private void sendCommitToOthers() throws RemoteException
	{
		for(int i = 0 ; i<this.others.length; i++){
			int yourPort = others[i];
			Registry re = LocateRegistry.getRegistry("localhost", yourPort);
			RemoteKeyValue st;
			try {
				st = (RemoteKeyValue) re.lookup(LOOKUPNAME);
				st.paxosHandleCommitSignalInLearner(req_a.id, req_a.type, req_a.args[0], req_a.args[1], myPort);
			} catch (NotBoundException e) {
				e.printStackTrace();
			}
		}
	}
	
	/** Checks its own ACKMap to see if it received acks from majority of
	 *  other servers for this request.*/
	private boolean checkMajorityAcks(Map<Integer, ACK> portAckMap)
	{
		int ackCounter = 0;
		for(int port : portAckMap.keySet()) 
		{
			if(portAckMap.get(port).ack == true){
				ackCounter++;
			}
		}
		return (ackCounter >= (NUMOFSERVERS+1)/2);
	}
	
	public void addInRequestList(UUID theReqId){
		//if(!reqList.contains(theReqId)) /**Sanity-check*/
		//	reqList.add(theReqId);
		this.firstACKMap.put(theReqId, Collections.synchronizedMap(new HashMap<Integer, ACK>()));
		this.secondACKMap.put(theReqId, Collections.synchronizedMap(new HashMap<Integer, ACK>()));
	}
	
	public void paxosHandlePrepareSignalInAcceptor(UUID transactionId, int n, int server) throws RemoteException
	{
		/** Simulate fault - the server will fail half the time. */
		if(Math.random() >= (1-FAILURE_PROBABILITY)) {
	        return;
	    }
		
		Registry reg = LocateRegistry.getRegistry("localhost", server);
		try{
			RemoteKeyValue stub = (RemoteKeyValue) reg.lookup(LOOKUPNAME);
			if(n < n_h)
			{
				/** Send prepare-reject */
				stub.paxosHandlePrepareAckInProposer(transactionId, false, n_a, req_a.id, req_a.type, req_a.args[0], req_a.args[1], myPort);
			}
			else
			{
				/** Send prepare-ok */
				n_h = n;
				if(req_a != null)
				{
					stub.paxosHandlePrepareAckInProposer(transactionId, true, n_a, req_a.id, req_a.type, req_a.args[0], req_a.args[1], myPort);
				}
				else
				{
					stub.paxosHandlePrepareAckWithNullReqInProposer(transactionId, true, myPort);
				}
			}
		}
		catch (Exception e) 
		{
			e.printStackTrace();
			System.out.print("Failed in paxosHandlePrepareSignalInAcceptor!");
		}
	}
	
	public void paxosHandlePrepareAckInProposer(
			UUID transactionId,
			boolean isPrepareOk,
			int n,
			UUID reqId,
			String req,	
			String key, 
			String value,
			int otherServerPort) throws RemoteException
	{
		if(!isPrepareOk)
		{
			addReqInMyFirstAckMap(transactionId, otherServerPort, false);
		}
		else
		{
			addReqInMyFirstAckMap(transactionId, otherServerPort, true);
			if(highestN_a < n)
			{
				highestN_a = n;
				String [] args = {key, value};
				highestReq_a = new Request(reqId, req, args);
			}
		}
		return;
	}
	
	public void paxosHandlePrepareAckWithNullReqInProposer(
			UUID transactionId,
			boolean isPrepareOk, 
			int otherServerPort) throws RemoteException
	{
		addReqInMyFirstAckMap(transactionId, otherServerPort, isPrepareOk);
	}

	public void paxosHandleAcceptSignalInAcceptor(
			UUID transactionId,
			int n, 
			UUID reqId, 
			String req, 
			String key, 
			String value,
			int server) throws RemoteException
	{
		try{
			Registry reg = LocateRegistry.getRegistry("localhost", server);
			RemoteKeyValue stub = (RemoteKeyValue) reg.lookup(LOOKUPNAME);
			if(n < n_h)
			{
				/** Send accept-reject */
				stub.paxosHandleAcceptAckInProposer(transactionId, false, myPort);
			}
			else
			{
				/** Send accept-ok */
				n_a = n;
				String [] args = {key, value};
				req_a = new Request(reqId, req, args);
				n_h = n;
				stub.paxosHandleAcceptAckInProposer(transactionId, true, myPort);
			}
		}
		catch (Exception e) 
		{
			e.printStackTrace();
			System.out.print("Failed in paxosHandleAcceptSignalInAcceptor!");
		}
	}

    public void paxosHandleAcceptAckInProposer(
    		UUID transactionId,
			boolean isAcceptOk, 
			int otherServerPort) throws RemoteException
	{
    	addReqInMySecondAckMap(transactionId, otherServerPort, isAcceptOk);
	}
	
    public void paxosHandleCommitSignalInLearner(
    		UUID reqId, 
			String req, 
			String key, 
			String value,
			int server) throws RemoteException
	{
    	String [] args = {key, value};
    	Request someReq = new Request(reqId, req, args);
    	this.processRequestUsingThreadPool(someReq);
	}
		
	public void addReqInMyFirstAckMap(UUID reqId, int serverPort, boolean isAccepted)
	{
		this.firstACKMap.get(reqId).put(serverPort, new ACK(isAccepted));  
	}
	
	public void addReqInMySecondAckMap(UUID reqId, int serverPort, boolean isAccepted)
	{
		this.secondACKMap.get(reqId).put(serverPort, new ACK(isAccepted));  
	}
}
	
	
